package com.levi.threadutils.service;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.levi.threadutils.UserThread;
import com.levi.threadutils.UserThread1;
import com.levi.threadutils.domain.User;
import com.levi.threadutils.mapper.UserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @description:
 * @author: shentong
 * @create: 2022-01-04 12:46
 **/
@Service
@Slf4j
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements UserService {

    /**
     * 多线程导入
     * @param
     * @return
     * @throws Exception
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void importData() throws Exception{
        long start_time = System.currentTimeMillis();
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //设置一个信号量为5的信号量，限制同时运行的线程数量最大为5
        Semaphore semaphore = new Semaphore(5);

        int num = 600;
        List<User> userList = getUserList();
        int rows = userList.size();
        int threadNum;
        if (rows % num>0) {
            threadNum =rows /num + 1;
        }else{
            threadNum = rows / num;
        }

        //设置一个倒计时门闩，用来处理主线程等待其他线程执行完成工作之后再运行
        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
        //创建一个定长的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);

        log.info("开始创建线程,数据总行数:{},线程数量:{}",rows,threadNum);

        List<Future<Integer>> futures = new ArrayList<>();
        int successCount = 0;
        for(int i = 1; i <= threadNum; i++){
            int startRow = (i-1)*num +1;
            int endRow = i*num;
            if (endRow>rows){
                endRow = rows;
            }
            log.info("开始执行线程方法,线程ID:<{}>,线程名称:<{}>",Thread.currentThread().getId(),Thread.currentThread().getName());
            Future<Integer> future = executorService.submit(new UserThread(userList,semaphore, startRow, endRow,this,countDownLatch));
            futures.add(future);
            log.info("结束线程执行方法,返回结果:<{}>,当前线程ID:<{}>,当前线程名称:<{}>", JSON.toJSONString(future),Thread.currentThread().getId(),Thread.currentThread().getName());
            //get方法中可以设置超时时间，即规定时间内没有返回结果，则继续运行
            //get方法是线程阻塞的，调用get方法会导致后续线程因主线程阻塞而没有创建，达不到效果。
            //successCount += future.get();
        }
        //主线程等待子线程完成任务,60秒还没执行完成就继续执行

        for(Future<Integer> future : futures){
            successCount += future.get();
        }
        //主线程等待子线程全部跑完才继续运行。可以设置等待时间，超时后继续执行。
        countDownLatch.await();
        //导入结束后关闭线程池
        executorService.shutdown();

        long end_time = System.currentTimeMillis();
        long use_time = (end_time - start_time) / 1000;
        log.info("用时{}秒,结束导入,共{}条数据，导入成功:{}", use_time, rows,successCount);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void importsingle() {
        long start_time = System.currentTimeMillis();
        List<User> userList = getUserList();
        this.saveBatch(userList);
        long end_time = System.currentTimeMillis();
        long use_time = (end_time - start_time) / 1000;
        log.info("用时{}秒,结束导入,共{}条数据", use_time, userList.size());
    }

    @Override
    public Integer datacount() {
        return this.count();
    }

    @Override
    public List<User> userList() {
        return this.list();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void importData1() throws Exception {
        long start_time = System.currentTimeMillis();

        List<User> userList = getUserList();
        int rows = userList.size();
        //根据线程数来分每个线程处理多少条数据
//        int num = 600;
//        //设置一个信号量为5的信号量，限制同时运行的线程数量最大为5
//        Semaphore semaphore = new Semaphore(5);
//        int threadNum;
//        if (rows % num>0) {
//            threadNum =rows /num + 1;
//        }else{
//            threadNum = rows / num;
//        }
        //创建一个定长的线程池，线程数量取cpu可用核心数，这样就不需要用信号量限制最大线程数量了
        int maxThreadNum = Runtime.getRuntime().availableProcessors() * 2;
        ExecutorService executorService = Executors.newFixedThreadPool(maxThreadNum);
        //设置一个倒计时门闩，用来处理主线程等待其他线程执行完成工作之后再运行
        CountDownLatch countDownLatch = new CountDownLatch(maxThreadNum);
        log.info("开始创建线程,数据总行数:{},线程数量:{}",rows,maxThreadNum);
        //根据线程数 任务平均分给每一个线程
        //每个线程要处理的任务数量
        //100个任务  3个线程  33 33 34
        int num1  = userList.size() / maxThreadNum;  //每个线程分的平均值
        int num2 = userList.size() % maxThreadNum;
        int lastNum = num1 + num2;//最后一个线程要处理的数据量
        List<Future<Integer>> futures = new ArrayList<>();
        int successCount = 0;
        for(int i = 1; i <= maxThreadNum; i++){
            int startRow = (i-1)*num1 +1;
            int endRow = i*num1;
            //最后一条线程 处理的数据多
            if (i == maxThreadNum){
                endRow = userList.size();
            }
            log.info("开始执行线程方法,线程ID:<{}>,线程名称:<{}>",Thread.currentThread().getId(),Thread.currentThread().getName());
            Future<Integer> future = executorService.submit(new UserThread1(userList, startRow, endRow,this,countDownLatch));
            futures.add(future);
            log.info("结束线程执行方法,返回结果:<{}>,当前线程ID:<{}>,当前线程名称:<{}>", JSON.toJSONString(future),Thread.currentThread().getId(),Thread.currentThread().getName());
            //get方法中可以设置超时时间，即规定时间内没有返回结果，则继续运行
            //get方法是线程阻塞的，调用get方法会导致后续线程因主线程阻塞而没有创建，达不到效果。
            //successCount += future.get();
        }
        //主线程等待子线程完成任务,60秒还没执行完成就继续执行

        for(Future<Integer> future : futures){
            successCount += future.get();
        }
        //主线程等待子线程全部跑完才继续运行。可以设置等待时间，超时后继续执行。
        countDownLatch.await();
        //导入结束后关闭线程池
        executorService.shutdown();

        long end_time = System.currentTimeMillis();
        long use_time = (end_time - start_time) / 1000;
        log.info("用时{}秒,结束导入,共{}条数据，导入成功:{}", use_time, rows,successCount);
    }

    private List<User> getUserList(){
        List<User> userList = new ArrayList<>();
        for (int i = 1; i <= 6000; i++) {
            User user = new User();
            user.setUsername("张三"+i);
            user.setPassword("123456");
            userList.add(user);
        }
        return userList;
    }

    @Transactional(rollbackFor = Exception.class)
    public Integer insertDB(List<User> userList){
        boolean result = this.saveBatch(userList);
        if (result) {
            return userList.size();
        }else{
            return  0;
        }
    }
}
